package com.mtf.mq.controller;

import com.alibaba.fastjson.JSON;
import com.mtf.common.enums.RocketMqEnum;
import com.mtf.entity.MtfOrder;
import com.mtf.handler.exception.RocketMqException;
import com.mtf.mq.config.RocketMQProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

/**
 * @Author matengfei
 * @Date 2024/3/13 20:36
 * @PackageName:com.mtf.mq.controller
 * @ClassName: TestRocketMqController
 * @Description: 测试RockerMq整合是否成功
 * @Version 1.0
 */
@RequestMapping("/mq")
@RestController
@Slf4j
public class TestRocketMqController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private RocketMQProperties rocketMQProperties;

    @GetMapping("/sendMsg")
    public String testSendMsg(@RequestParam("msg") String msg){
        Message<String> message = MessageBuilder.withPayload(msg).build();
        rocketMQTemplate.send("idempotent_topic",message);
        return msg;
    }

    /**
     * @Description: 发送分区顺序消息（也叫局部顺序消息）
     * @return java.lang.String
     */
    @GetMapping("/sendPartitionOrderMessage")
    public String sendPartitionOrderMessage() throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        //创建生产者
        DefaultMQProducer producer = new DefaultMQProducer(RocketMqEnum.TEST_PRODUCER_GROUP.getGroupName());
        producer.setNamesrvAddr(rocketMQProperties.getNamesrvaddr());
        MtfOrder order1 = new MtfOrder(1,"A001",10);
        MtfOrder order2 = new MtfOrder(2,"A002",20);
        MtfOrder order3 = new MtfOrder(3,"A003",30);
        List<MtfOrder> orderList = Arrays.asList(order1, order2, order3);
        producer.start();
        //循环发送,模拟顺序消息
        for (MtfOrder order : orderList) {
            //创建消息
            org.apache.rocketmq.common.message.Message message = new org.apache.rocketmq.common.message.Message();
            message.setTopic(RocketMqEnum.TEST_PRODUCER_GROUP.getTopicName());
            message.setTags(RocketMqEnum.TEST_PRODUCER_GROUP.getTagName());
            String jsonString = JSON.toJSONString(order);
            message.setBody(jsonString.getBytes());
            //设置分区键，保证当前消费者发送的多条消息被发送到同一个队列
            message.setKeys(order.getOrderName());
            producer.send(message);
        }
        return "顺序消息发送结束！";
    }

    /**
     * @Description: 发送全局顺序消息
     * @param msg
     * @return java.lang.String
     */
    @GetMapping("/sendGlobalOrderMessage")
    public String sendGlobalOrderMessage(@RequestParam("msg") String msg) throws MQClientException {
        //创建生产者
        DefaultMQProducer producer = new DefaultMQProducer(RocketMqEnum.TEST_PRODUCER_GROUP.getGroupName());
        producer.setNamesrvAddr(rocketMQProperties.getNamesrvaddr());
        producer.start();
        //创建topic,并指定topic下的队列数量（其实没有必要,分区顺序消息足够,这样的全局只会限制mq的优势）
        //producer.createTopic("与业务相关联的分区键","topic",1);
        //创建消息
        org.apache.rocketmq.common.message.Message message = new org.apache.rocketmq.common.message.Message();
        message.setTopic(null);
        message.setTags(null);
        message.setBody(null);
        //设置分区键，保证当前消费者发送的多条消息被发送到同一个队列
        message.setKeys("与业务相关联的分区键");
        try {
            producer.send(message);
        } catch (Exception e) {
            log.info("消息发送异常！");
        }
        return msg;
    }
}
